package com.myflink.day04;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author Shelly An
 * @create 2020/9/19 11:23
 * flink 事件驱动（被动）  有数据才计算
 * spark 主动 到时间就计算
 * kafka 是事件驱动
 */
public class Window_CountWindow {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);


        DataStreamSource<String> socketDS = env.socketTextStream("master", 4444);

        //DataStream可以直接调用开窗的方法，但是都带“all”，这种情况下所有数据不分组，都在窗口里。开发中不建议使用
        //一般先分组，再使用开窗方法
        KeyedStream<Tuple2<String, Integer>, String> dataKS = socketDS.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {
                return Tuple2.of(value, 1);
            }
        }).keyBy(data -> data.f0);

        //根据本组内达到多少条 而非整个流 -->因为是keyBy之后开的窗
        dataKS
                //滚动窗口: 一个参数
                .countWindow(3)
                .sum(1)
                .print("cw1");
        dataKS
                //滑动窗口：两个参数  每经过一次步长会触发一次滑动窗口的关闭  窗口长度/步长=数据被计算多少次
                .countWindow(3, 2)
                .sum(1)
                .print("cw2");


        env.execute();
    }
}
